package chapter07

object Demo18_ComplexWordCount {
  def main(args: Array[String]): Unit = {
    val tupleList: List[(String , Int)] = List(
      ("hello word" , 2) ,
      ("hello scala" , 3),
      ("hello hadoop" , 4),
      ("hello flink" , 3)
    )
    //思路一 ： 展开为普通版本
    //trim
    val newStringList  = tupleList.map(
      kv => {
        (kv._1.trim +" " ) * kv._2
      }
    )
    //普通版本解决就行
    val wordCountList: List[(String , Int)] = newStringList
      .flatMap(_.split( " "))
      .groupBy(word => word)
      .map( kv => (kv._1 ,kv._2.size))
      .toList
      .sortBy(_._2)(Ordering[Int].reverse)
      .take(3)

    println(wordCountList)

    //直接基于预统计的结果进行转换
    //1、将字符串打散为单词并结合包装成元组
    val preCountList: List[(String ,Int)] = tupleList.flatMap(
      tuple => {
        val strings: Array[String] = tuple._1.split(" ")
        strings.map(word => (word , tuple._2))
      }
    )
    println(preCountList)
    //2、对二元组按照单词进项分组
    val preCountMap: Map[String,List[(String ,Int)]] = preCountList.groupBy(_._1)
    println(preCountMap)
    //3、叠加每个单词预统计的个数值
    val countMap = preCountMap.mapValues(
      tupleList => tupleList.map(_._2).sum
    )
    println(countMap)
    //4、转换成list，排序取三
    val countList = countMap.toList
      .sortWith(_._2 > _._2)
      .take(3)
    println(countList)
  }
}
